package com.wsx.ones.monitor.core.exec.pull;

import com.google.common.util.concurrent.*;
import com.wsx.ones.monitor.core.executer.RpcExecuter;
import com.wsx.ones.monitor.core.model.RpcRequest;
import com.wsx.ones.monitor.core.model.RpcResponse;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

/**
 * 任务的执行操作
 * 该类是核心的操作这里做详细的讲解
 * 接受的参数中必须有ChannelHandlerContext，该参数可以将处理的结果反写给客户端
 * Created by wangshuaixin on 17/2/7.
 */
public class PullExecuter {

    //定义线程池的处理框架
    private static ListeningExecutorService threadPool;

    public static void submit(PullMessageHandleTask handleTask, RpcRequest rpcRequest,
                              RpcResponse rpcResponse, ChannelHandlerContext ctx) {
        //线程池的初始化操作
        if (null == threadPool) {
            synchronized (PullExecuter.class) {
                if (null == threadPool) {
                    threadPool = MoreExecutors.listeningDecorator(RpcExecuter.getExecuter(10, 50, -1));
                }
            }
        }

        //采用guava的多线程处理框架
        ListenableFuture<Boolean> future = threadPool.submit(handleTask);
        Futures.addCallback(future, new FutureCallback<Boolean>() {
            //操作成功后的动作
            @Override
            public void onSuccess(Boolean result) {
                //处理成功后将结果反写给客户端
                ctx.writeAndFlush(rpcResponse).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        // TODO log
                    }
                });
            }

            @Override
            public void onFailure(Throwable t) {

            }
        }, threadPool);
    }

}
